package com.dky.satell.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Message;
import com.dky.satell.common.KafkaKeyUtil;
import com.dky.satell.entity.MessageConfig;
import com.dky.satell.kafka.model.MessageWrapper;
import com.dky.satell.kafka.model.OfflineFileWarpper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * kafka生产者
 *
 * @author lihaobin
 * @version 1.0
 * @date 2021/4/20 10:11
 */
@Slf4j
@Service
public class KafkaProducer {

    @Value("${kafka.transfer.topicPrefix}")
    private String topicPrefix;

    @Value("${kafka.transfer.topic.file}")
    private String fileTopic;

    @Value("${kafka.transfer.topic.messageConfig}")
    private String messageConfigTopic;

    @Value("${kafka.transfer.topic.message}")
    private String messageTopic;

    @Resource
    private KafkaTemplate kafkaTemplate;

    public void sendFile(OfflineFileWarpper warpper) {
        log.info("KafkaProducer sendFile: [{}]", JSON.toJSONString(warpper));
//        ProducerRecord record = new ProducerRecord(topicPrefix + fileTopic, KafkaKeyUtil.keyByZone(warpper.getOfflineFile().getRegion()), JSON.toJSONString(warpper));
//        kafkaTemplate.send(record);
    }

    public void sendMessageConfig(MessageConfig messageConfig) {
        log.info("KafkaProducer sendMessageConfig: [{}]", JSON.toJSONString(messageConfig));
//        ProducerRecord record = new ProducerRecord(topicPrefix + messageConfigTopic, KafkaKeyUtil.keyByZone(messageConfig.getRegion()), JSON.toJSONString(messageConfig));
//        kafkaTemplate.send(record);
    }

    public void sendMessage(MessageWrapper warpper) {
        log.info("KafkaProducer sendMessage: [{}]", JSON.toJSONString(warpper));
        ProducerRecord record = new ProducerRecord(topicPrefix + messageTopic, KafkaKeyUtil.keyByZone(warpper.getMessageConfig().getRegion()), JSON.toJSONString(warpper));
        kafkaTemplate.send(record);
    }

    //异步发送消息回调
    public void sendAsyncMessage() {
        Map<String, String> map = new HashMap<>();
        map.put("Aj", "90");
        ListenableFuture<SendResult<String, Message>> topic = kafkaTemplate.send("topic", JSON.toJSONString(map));
        topic.addCallback(new ListenableFutureCallback<SendResult<String, Message>>() {
            //发送消息失败回调
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }

            //发送消息成功回调
            @Override
            public void onSuccess(SendResult<String, Message> result) {
                System.out.println(result.getProducerRecord());
                System.err.println(result.getRecordMetadata());
            }
        });
    }

}
